var zookeeper = require('node-zookeeper-client');
var amqp = require('amqplib/callback_api');

var __zkNode = '/myrpc/server/currid';
var __zkConfig = 'localhost:2181';
var __zkBaseId = '1';

var __rabbitmqConfig = 'amqp://localhost';
var __qRequest = 'request_to_{type}';
var __qReply = 'reply_to_{appId}';


// app.handle and app.type
function AppWrapper(app) {
	if (this instanceof AppWrapper) {
		if (!app.type) {
			throw new Error('app must have a type:string');
		}
		if (!app.handle || Object.prototype.toString.call(app.handle) !== '[object Function]') {
			throw new Error('app must have a handle:function');
		}
		this.app = app;
		this.type = app.type;
		return this;
	}
	return new AppWrapper(app);
}
module.exports = AppWrapper;


AppWrapper.prototype.run = function(cb) {
	var self = this;
	self.prepare(cb);
}

AppWrapper.prototype.getId = function() {}
AppWrapper.prototype.nextMsgId = function() {
	var self = this;
	if (!self.msgId) self.msgId = 1;
	else self.msgId++;
	return self.msgId + '';
}

AppWrapper.prototype.rpc = function(type, args, callback) {
	var self = this;
	// send to mq
	var qSend = __qRequest.replace('{type}', type);
	self.__mqCh.assertQueue(qSend, {
		durable: true
	}, function(err, q) {
		var msgId = self.nextMsgId();
		self.__mqCh.sendToQueue(qSend, new Buffer(args), {
			correlationId: msgId,
			replyTo: self.appId + '',
			persisitent: true,
		});

		// wait for reply then callback
		self.registerRpcCallback(msgId, callback);
	});
};

AppWrapper.prototype.registerRpcCallback = function(msgId, callback) {
	var self = this;
	if (!self.rpcCallback) {
		self.rpcCallback = {};
	}
	self.rpcCallback[msgId] = callback;
}

AppWrapper.prototype.handleRequest = function(args, sendBack) {
	this.app.handle(args, sendBack);
};

AppWrapper.prototype.receiveReply = function(msgId, content) {
	var self = this;
	if (!self.rpcCallback) {
		self.rpcCallback = {};
	}
	var cb = self.rpcCallback[msgId];
	if (!cb) {
		console.error('no callback with msgId = %s', msgId);
	} else {
		cb(content);
	}
}

/**
 * connect to zookeeper
 * connect to rabbitMQ
 */
AppWrapper.prototype.prepare = function(cb) {
	var self = this;
	var zkClient = zookeeper.createClient(__zkConfig);
	zkClient.connect();
	zkClient.once('connected', function() {
		console.log('connected to zk server');
		__prepareZk(zkClient, function(err, appId) {
			if (err) {
				console.log(err);
				throw new Error('zookeeper fail');
			} else {
				console.log('prepare zk done, appId=' + appId);
				self.appId = appId;
				//------------------------
				__prepareMq(self, function() {
					console.log('prepare mq done');
					cb();
				});
			}
		});
	});
}


function __prepareZk(zk, cb) {
	zk.exists(__zkNode, function(err, stat) {
		if (err) return cb(err);
		if (stat) { // exists
			console.log('exists node %s', __zkNode);
			zk.getData(__zkNode, null, function(err, data, stat) {
				console.log('getData node %s', __zkNode);
				if (err) return cb(err);
				// console.log('Got data: %s', data.toString('utf8'));
				var preId = Number(data);
				var myId = preId + 1;
				zk.setData(__zkNode, new Buffer(myId + ''), -1, function(err, stat) {
					cb(err, myId);
				})
			});
		} else { // not exists
			console.log('not exists node %s', __zkNode);
			zk.mkdirp(__zkNode, new Buffer(__zkBaseId), function(err, path) {
				console.log('create node %s', __zkNode);
				cb(err, __zkBaseId);
			});
		}
	});
}


function __prepareMq(self, cb) {
	amqp.connect(__rabbitmqConfig, function(err, conn) {
		conn.createChannel(function(err, ch) {
			console.log('connect to mq');
			self.__mqCh = ch;
			// create many queue
			// qRequst = request_{mytpye}
			// qReply = myidis.{myid}

			// get reply from this
			var qReply = __qReply.replace('{appId}', self.appId);
			ch.assertQueue(qReply, {
				exclusive: true
			}, function(err, q) {
				console.log('create queue %s', qReply);
				ch.consume(qReply, function(msg) {
					ch.ack(msg); // ack
					console.log('receive reply: %s', msg.content);
					var msgId = msg.properties.correlationId;
					self.receiveReply(msgId, msg.content);
				});


				// get request from qRequst
				var qRequst = __qRequest.replace('{type}', self.type);
				console.log('create queue %s', qRequst);
				ch.assertQueue(qRequst, {
					durable: true
				});
				ch.prefetch(1);
				ch.consume(qRequst, function(msg) {
					console.log('receive request:%s', msg.content);
					ch.ack(msg);

					// handle and reply 
					self.handleRequest(msg.content, function(result) {
						var replyTo = msg.properties.replyTo;
						var replyQid = __qReply.replace('{appId}', replyTo);
						console.log('handle done, replyQid=%s, result=%s', replyQid, result);
						ch.sendToQueue(replyQid, new Buffer(result), {
							correlationId: msg.properties.correlationId
						});
					});
				}, {
					noAck: false
				});


				cb();
			});
		});
	});
}